package com.atguigu.gmall.realtime.app.newapp

import com.atguigu.gmall.realtime.app.newapp.DwRefundApp.sparkConf
import com.atguigu.gmall.realtime.bean.{OrderRefundInfo, PageLog, SkuVisitLog}
import com.atguigu.gmall.realtime.handler.{CommonApp, DimHandler, EsHandler}
import com.atguigu.gmall.realtime.util.MyBeanUtil
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

import java.text.SimpleDateFormat
import java.time.LocalDate
import java.util.Date

//  用户 sku 关联
object DwSkuVisitApp extends CommonApp {

  def main(args: Array[String]): Unit = {

    val groupId: String = "dw_sku_visit_app"
    val topic: String = "DWD_PAGE_LOG"
    val ssc: StreamingContext = initSsc(groupId, "local[4]", 5)

    val dStream: DStream[PageLog] = getDstream[PageLog](ssc, topic, groupId)


    val filteredDstream: DStream[PageLog] = dStream.filter { pageLog =>
      pageLog.page_id == "good_detail"
    }

    val skuVisitDstream: DStream[SkuVisitLog] = filteredDstream.map { pageLog =>
      val skuVisitLog = SkuVisitLog()
      MyBeanUtil.copyProperties(pageLog, skuVisitLog)
      val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
      val dtHour: String = dateFormat.format(new Date(skuVisitLog.ts))
      val dtHourArr: Array[String] = dtHour.split(" ")
      skuVisitLog.dt = dtHourArr(0)
      skuVisitLog.hr = dtHourArr(1)


      skuVisitLog.sku_id = pageLog.page_item.toLong
      //幂等性处理
      skuVisitLog.log_id= s"${skuVisitLog.user_id}_${skuVisitLog.sku_id}_${skuVisitLog.ts / 1000}"


      skuVisitLog
    }


    val dataWithUDstream: DStream[SkuVisitLog] = DimHandler.joinUser[SkuVisitLog](skuVisitDstream,
      { skuVisitLog: SkuVisitLog => skuVisitLog.user_id.toString }
    )

    val dataWithSkuDStream: DStream[SkuVisitLog] = DimHandler.joinSku[SkuVisitLog](dataWithUDstream,
      { skuVisitLog: SkuVisitLog => skuVisitLog.sku_id.toString }
    )

    val dataWithProvinceDStream: DStream[SkuVisitLog] = DimHandler.joinProvince[SkuVisitLog](dataWithSkuDStream,
      { skuVisitLog: SkuVisitLog => skuVisitLog.province_id }
    )
    dataWithProvinceDStream.cache()
    dataWithProvinceDStream.print(1000)




    EsHandler.saveDStreamToEs[SkuVisitLog](dataWithProvinceDStream,
      "gmall2022_sku_visit_log_",
      "log_id",
      {Unit=>this.commitAllOffsetKafka()}
    )

    ssc.start()
    ssc.awaitTermination()

  }

}
